import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.annotation.{DataTypeHint, FunctionHint}
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.functions.TableFunction
import org.apache.flink.types.Row

/**
 * Author: jiwei01
 * Date: 2022/6/21 11:47
 * Package: 
 * Description:
 */
@FunctionHint(output = new DataTypeHint("ROW<word STRING, length INT>"))
class SplitFunction extends TableFunction[Row] {

  def eval(str: String): Unit = {
    // use collect(...) to emit a row
    str.split(" ").foreach(s => collect(Row.of(s, Int.box(s.length))))
  }
}

object SQLSample extends App {
  val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
  env.setParallelism(1)
  val tEnv = StreamTableEnvironment.create(env)

  val words = List("Hello World", "Hello Scala")

  val inputTable: Table = tEnv.fromDataStream(
    env.fromCollection(words).flatMap(_.toList))
  tEnv.createTemporaryView("input_table", inputTable)
  val outputTable: Table = tEnv.sqlQuery(
    """
      |SELECT f0, COUNT(*) FROM input_table
      |GROUP BY f0
      |""".stripMargin)

  tEnv.toChangelogStream(outputTable).print()

  env.execute()
}
